package user.service.Impl;

import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import user.contstant.Constants;
import user.service.IUserService;
import user.vo.Response;
import user.vo.User;

import java.util.ArrayList;
import java.util.List;


/**
 * 创建用户服务实现
 */
@Slf4j
@Service
public class UserServiceImpl implements IUserService {

    /** HBase客户端*/
    @Autowired
    private HbaseTemplate hbaseTemplate;
    /**redis客户端*/
    @Autowired
    private StringRedisTemplate redisTemplate;

    @Override
    public Response createUser(User user) throws Exception {

        //定义列族B,基本信息
        byte[] FAMILY_B = Constants.UserTable.FAMILY_B.getBytes();
        byte[] Name = Constants.UserTable.NAME.getBytes();
        byte[] AGE = Constants.UserTable.AGE.getBytes();
        byte[] SEX = Constants.UserTable.SEX.getBytes();
        //定义列族o，额外信息
        byte[] FAMILY_O = Constants.UserTable.FAMILY_O.getBytes();
        byte[] PHONE = Constants.UserTable.PHONE.getBytes();
        byte[] ADDRESS = Constants.UserTable.ADDRESS.getBytes();

        /** 当前用户总数*/
        Long curCount = redisTemplate.opsForValue().increment(Constants.USE_COUNT_REDIS_KEY,1);

        /**生成userId*/
        Long userId = getUserId(curCount);

        List<Mutation> datas = new ArrayList<>();

        /**写入HBase*/
        Put put = new Put(Bytes.toBytes(userId));
        /** 列族B*/
        put.addColumn(FAMILY_B,Name,Bytes.toBytes(user.getBaseInfo().getName()));
        put.addColumn(FAMILY_B,AGE,Bytes.toBytes(user.getBaseInfo().getAge()));
        put.addColumn(FAMILY_B,SEX,Bytes.toBytes(user.getBaseInfo().getSex()));

        /**列族O*/
        put.addColumn(FAMILY_O,PHONE,Bytes.toBytes(user.getOtherInfo().getPhone()));
        put.addColumn(FAMILY_O,ADDRESS,Bytes.toBytes(user.getOtherInfo().getAddress()));

        /**写入HBase*/
        datas.add(put);
        hbaseTemplate.saveOrUpdate(Constants.UserTable.TABLE_NAME, (Mutation) datas);

        user.setId(userId);
        return new Response(user);
    }

    /**
     * 生成userId
     * @param prefix 当前用户数
     * @return 用户id
     */
    private Long getUserId(Long prefix){
        String suffix = RandomStringUtils.randomNumeric(5);
        return Long.valueOf(prefix+suffix);
    }
}
